package com.lagou.kafka.demo.controller;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

/**
 * @param
 * @author oyz
 * @version 1.0
 * @description: TODO 控制层(同步发送)
 * @date 2022/3/25 21:28
 */
@RestController
public class KafkaAsyncProducerController {

    @Autowired
    private KafkaTemplate<Integer,String> template;

    @RequestMapping("/send/async/{message}")
    public String send(@PathVariable String message){
        ListenableFuture<SendResult<Integer, String>> result = template.send("topic-spring-01", 0, 1, message);
        result.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("消息发送失败"+throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<Integer, String> integerStringSendResult) {
                final RecordMetadata metadata = integerStringSendResult.getRecordMetadata();
                System.out.println(metadata.partition()+"\t"+metadata.topic()+"\t"+metadata.offset());
            }
        });


        return "success";

    }


}
